package com.chb.flink.time

import com.chb.flink.source.{MyCustomerSource, StationLog}
import org.apache.flink.api.common.functions.ReduceFunction
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.watermark.Watermark
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

/**
 * 每隔5秒统计一下最近10秒内，每个基站中通话时间最长的一次通话发生的时间还有，
 * 主叫号码，被叫号码，通话时长，并且还得告诉我们当前发生的时间范围（10秒）
 */
object MaxCallTime2 {

    def main(args: Array[String]): Unit = {
        val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
        import org.apache.flink.streaming.api.scala._
        streamEnv.setParallelism(1)

        //设置时间语义
        streamEnv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
        streamEnv.getConfig.setAutoWatermarkInterval(1000L) // 指定注入WaterMark的周期间隔

        // 接入数据源, 数据是乱序
        val data = streamEnv.addSource(new MyCustomerSource)
            .filter(_.callType == "success")
            //            .assignAscendingTimestamps(_.callTime) // 有序数据引入watermark, 通过参数指定eventTime
//            .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[StationLog](Time.seconds(3)) {
//                override def extractTimestamp(t: StationLog): Long = {
//                    // 指定eventTime的字段
//                    t.callTime
//                }
//            })
                .assignTimestampsAndWatermarks(new MyCustomPeriodicWater(3000L))

        //分组、开窗
        data.filter(_.callType.equals("success")).keyBy(_.sid)
            .timeWindow(Time.seconds(10), Time.seconds(5))
            .reduce(new MyReduceFunction(), new ReturnMaxTimeWindowFunction)
            .print()

        streamEnv.execute()

    }

    // 自定义的
    class MyCustomPeriodicWater(delay: Long) extends AssignerWithPeriodicWatermarks[StationLog]{
        var maxEventTime: Long = 0
        override def getCurrentWatermark: Watermark = {
            new Watermark(maxEventTime - delay)
        }

        override def extractTimestamp(t: StationLog, l: Long): Long = {
            maxEventTime = maxEventTime.max(t.callTime)
            // 指定EventTime
            t.callTime
        }
    }

    /**
     * 聚合函数
     */
    class MyReduceFunction() extends ReduceFunction[StationLog] { //增量聚合
        override def reduce(value1: StationLog, value2: StationLog): StationLog = {
            if (value1.duration > value2.duration) value1 else value2
        }
    }

}